package com.zha;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.IFile;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.File;
import java.io.IOException;
//提取一个网页里的图片，例如<img src="http://t1.27270.com/uploads/tu/201812/83/667f2957af.jpg" width="190" height="280" 把网址取出来
//此程序不需要reduce部分
public class YeMianTiQu {
    public static class MapYeMianTiQu extends Mapper<LongWritable, Text,Text, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String s = value.toString();
            //string的包含（continue）
            if (s.contains("img src=")){
                //进行切分
                String[] line = s.split("img src=\"");//   \" 是转义操作
                for (String s1 : line) {
                    //判断是否有.jpg的文件，如果有再进行切分
                    if (s1.contains(".jpg")){
                        //取数组的第0位，也就是前面的网址
                        String s2 = s1.split("\" width")[0];
                        context.write(new Text(s2),NullWritable.get());
                    }else {
                        continue;//continue只能在for循环里使用，而return在if和for里都能用，所以这里也可写return
                    }
                }
            }else {
                return;
            }
        }

    }
//    public static class ReduceYeMianTiQu extends Reducer<Text,IntWritable,Text,IntWritable>{
//        @Override
//        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
//           int a = 0;
//            for (IntWritable value : values){
//                a = a + value.get();
//            }
//            context.write(key,new IntWritable(a));
//    }

        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            Configuration conf =new Configuration();
            Job job = new Job();

            job.setJarByClass(YeMianTiQu.class);
            job.setMapperClass(MapYeMianTiQu.class);
//            job.setReducerClass(ReduceYeMianTiQu.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);

//            job.setOutputKeyClass(Text.class);
//            job.setOutputValueClass(IntWritable.class);

            File file = new File("d:/file/output");
            if (file.exists()) {
                FileUtil.fullyDelete(file);
            }
            FileInputFormat.addInputPath(job,new Path("d:/file/input"));
            FileOutputFormat.setOutputPath(job,new Path("d:/file/output"));

            job.setNumReduceTasks(1);
            System.exit(job.waitForCompletion(true)?0:1);
        }
    }

